package com.raymond.queue.handler;

import com.raymond.queue.Production;
import com.raymond.queue.QueueManager;
import com.raymond.queue.common.msg.request.ProductionMsgRequest;
import com.raymond.queue.common.msg.response.ProductionMsgResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 生产者请求处理
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2021-07-14 11:04
 */
public class ProductionMsgRequestHandler extends SimpleChannelInboundHandler<ProductionMsgRequest> {
    private final static Logger logger = LoggerFactory.getLogger(ProductionMsgRequestHandler.class);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ProductionMsgRequest msg) throws Exception {
        try {
            logger.info("生产消息,topic:{}", msg.getTopic());
            Production production = QueueManager.getProduction(msg.getTopic());
            if (production == null) {
                ctx.writeAndFlush(new ProductionMsgResponse(1, msg.getSequenceId()));
            } else {
                production.put(msg.getData());
                ctx.writeAndFlush(new ProductionMsgResponse(0, msg.getSequenceId()));
            }
        } catch (Exception e) {
            ctx.writeAndFlush(new ProductionMsgResponse(1, msg.getSequenceId()));
            logger.error("生产数据异常,客户端地址:{},topic:{},data:{}", ctx.channel().remoteAddress(), msg.getTopic(), msg.getData(), e);
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("ProductionMsgRequestHandler Established connection with the remote client.");

        // do something

        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("ProductionMsgRequestHandler Disconnected with the remote client.");

        // do something

        ctx.fireChannelInactive();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}
